# Week 3 | 第2课：LangGraph 基础

**课程编号**: 3.2
**时长**: 45 分钟
**前置**: 3.1 为什么需要状态机

---

## 学习目标

- 掌握 LangGraph 的四个核心概念：State、Node、Edge、Conditional Edge
- 能够搭建最简单的线性图（Start → Node1 → Node2 → End）
- 能够添加条件分支（if-else 路由）
- 理解状态在节点之间如何传递和更新

---

## 1. 环境准备

```bash
# 创建虚拟环境（推荐）
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

# 安装 LangGraph
pip install langgraph langchain-openai

# 设置 API Key
export OPENAI_API_KEY="your-api-key-here"
# Windows PowerShell: $env:OPENAI_API_KEY="your-api-key-here"
```

---

## 2. 核心概念

LangGraph 只有四个核心概念，理解了就能写出任何流程图：

| 概念 | 说明 | 类比 |
|------|------|------|
| **State** | 贯穿整个图的数据结构 | 一张在整个流程中传递的"工单" |
| **Node** | 处理状态的函数 | 工单在每个环节的处理操作 |
| **Edge** | 节点之间的固定连线 | "做完 A 就做 B" |
| **Conditional Edge** | 根据状态决定下一步走哪条路 | "如果 X 做 A，否则做 B" |

### 2.1 State（状态）

State 是整个流程中传递的数据。它定义了你的"工单"上有哪些字段。

```python
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, END
import operator

# 定义状态：一个简单的字典结构
class AgentState(TypedDict):
    # 用户的输入问题
    question: str
    # 分类结果
    category: str
    # 处理后的回复
    reply: str
    # 满意度检查结果
    satisfied: bool
    # 循环计数器（防止死循环）
    loop_count: int
```

**关键点**：State 中的字段会在节点之间传递和更新。每个节点接收当前 State，处理后返回更新后的 State。

### 2.2 Node（节点）

Node 是一个普通 Python 函数，它接收 State 作为参数，返回更新后的 State。

```python
def classify_node(state: AgentState) -> dict:
    """分类节点：判断用户问题属于哪个类别"""
    question = state["question"]

    # 这里简化处理，实际应该调用 LLM
    if "账单" in question or "付款" in question:
        category = "billing"
    elif "故障" in question or "不能用" in question:
        category = "tech"
    else:
        category = "support"

    # 返回更新后的状态（只返回需要更新的字段）
    return {"category": category, "loop_count": state["loop_count"] + 1}
```

### 2.3 Edge（边）

Edge 连接两个节点，表示固定的执行顺序。

```python
# 添加边：classify_node 完成后，固定路由到 billing_node
workflow.add_edge("classify", "billing")
```

### 2.4 Conditional Edge（条件边）

Conditional Edge 根据当前状态动态决定下一个节点。

```python
def route_by_category(state: AgentState) -> str:
    """路由函数：根据分类结果决定走哪条路"""
    if state["category"] == "billing":
        return "billing"
    elif state["category"] == "tech":
        return "tech"
    else:
        return "support"

# 添加条件边
workflow.add_conditional_edges(
    "classify",        # 从哪个节点出发
    route_by_category, # 路由函数
    {                   # 路由函数的返回值映射到哪个节点
        "billing": "billing",
        "tech": "tech",
        "support": "support",
    }
)
```

---

## 3. 从零搭建第一个图

### 3.1 最简单的线性图

我们先搭建一个没有任何分支的最简图：Start → Node1 → Node2 → End。

```python
"""
lesson3_2_simple_graph.py
最简 LangGraph：Start → Greet → Answer → End
"""

from typing import TypedDict
from langgraph.graph import StateGraph, END

# ========== 1. 定义 State ==========
class SimpleState(TypedDict):
    user_name: str
    greeting: str
    answer: str

# ========== 2. 定义 Nodes ==========
def greet_node(state: SimpleState) -> dict:
    """节点1：打招呼"""
    name = state.get("user_name", "用户")
    greeting = f"你好，{name}！"
    print(f"[Greet] {greeting}")
    return {"greeting": greeting}

def answer_node(state: SimpleState) -> dict:
    """节点2：回答问题"""
    answer = "这是你的答案：LangGraph 让流程可控！"
    print(f"[Answer] {answer}")
    return {"answer": answer}

# ========== 3. 构建 Graph ==========
# 创建图实例，指定 State 类型
graph_builder = StateGraph(SimpleState)

# 添加节点
graph_builder.add_node("greet", greet_node)
graph_builder.add_node("answer", answer_node)

# 设置入口点（从 START 到 greet 节点）
graph_builder.set_entry_point("greet")

# 添加固定边：greet → answer
graph_builder.add_edge("greet", "answer")

# 添加固定边：answer → END
graph_builder.add_edge("answer", END)

# 编译图
simple_graph = graph_builder.compile()

# ========== 4. 运行图 ==========
if __name__ == "__main__":
    result = simple_graph.invoke({"user_name": "张三"})
    print("\n===== 最终结果 =====")
    print(f"greeting: {result['greeting']}")
    print(f"answer: {result['answer']}")
```

**运行结果：**
```
[Greet] 你好，张三！
[Answer] 这是你的答案：LangGraph 让流程可控！

===== 最终结果 =====
greeting: 你好，张三！
answer: 这是你的答案：LangGraph 让流程可控！
```

**流程拆解：**
1. `set_entry_point("greet")` 告诉图从 greet 节点开始
2. `add_edge("greet", "answer")` 表示 greet 完成后固定执行 answer
3. `add_edge("answer", END)` 表示 answer 完成后结束

### 3.2 可视化你的图

LangGraph 内置了可视化功能：

```python
# 生成图的 PNG 可视化
from IPython.display import Image, display

# 在 Jupyter 中直接显示
display(Image(simple_graph.get_graph().draw_mermaid_png()))

# 或者生成 mermaid 文本查看
print(simple_graph.get_graph().draw_mermaid())
```

生成的 mermaid 图长这样：
```mermaid
graph TD
    __start__ --> greet
    greet --> answer
    answer --> __end__
```

---

## 4. 添加条件分支

现在我们在上面的基础上加一个条件分支：根据用户类型走不同的处理路径。

```python
"""
lesson3_2_conditional_graph.py
带条件分支的 LangGraph
Start → Classify → (VIP?) → VIP处理 / 普通处理 → End
"""

from typing import TypedDict
from langgraph.graph import StateGraph, END

# ========== 1. 定义 State ==========
class CustomerState(TypedDict):
    user_name: str
    user_type: str       # "vip" 或 "normal"
    reply: str

# ========== 2. 定义 Nodes ==========
def classify_node(state: CustomerState) -> dict:
    """分类：判断用户类型"""
    # 简化：名字里带 VIP 的就是 VIP 用户
    name = state.get("user_name", "")
    user_type = "vip" if "VIP" in name else "normal"
    print(f"[Classify] 用户类型: {user_type}")
    return {"user_type": user_type}

def vip_node(state: CustomerState) -> dict:
    """VIP 处理"""
    reply = f"尊贵的 VIP 用户 {state['user_name']}，您的专属客服已上线！"
    print(f"[VIP] {reply}")
    return {"reply": reply}

def normal_node(state: CustomerState) -> dict:
    """普通用户处理"""
    reply = f"您好 {state['user_name']}，请问有什么可以帮您？"
    print(f"[Normal] {reply}")
    return {"reply": reply}

# ========== 3. 路由函数 ==========
def route_by_type(state: CustomerState) -> str:
    """根据用户类型路由"""
    return state["user_type"]  # 返回 "vip" 或 "normal"

# ========== 4. 构建 Graph ==========
builder = StateGraph(CustomerState)

# 添加所有节点
builder.add_node("classify", classify_node)
builder.add_node("vip", vip_node)
builder.add_node("normal", normal_node)

# 入口点
builder.set_entry_point("classify")

# 条件边：classify 完成后，根据 user_type 路由
builder.add_conditional_edges(
    "classify",
    route_by_type,
    {
        "vip": "vip",
        "normal": "normal",
    }
)

# 两条路径都通向 END
builder.add_edge("vip", END)
builder.add_edge("normal", END)

# 编译
customer_graph = builder.compile()

# ========== 5. 运行测试 ==========
if __name__ == "__main__":
    print("=== 测试 VIP 用户 ===")
    result1 = customer_graph.invoke({"user_name": "VIP_张三"})
    print(f"结果: {result1['reply']}\n")

    print("=== 测试普通用户 ===")
    result2 = customer_graph.invoke({"user_name": "李四"})
    print(f"结果: {result2['reply']}")
```

**运行结果：**
```
=== 测试 VIP 用户 ===
[Classify] 用户类型: vip
[VIP] 尊贵的 VIP 用户 VIP_张三，您的专属客服已上线！
结果: 尊贵的 VIP 用户 VIP_张三，您的专属客服已上线！

=== 测试普通用户 ===
[Classify] 用户类型: normal
[Normal] 您好 李四，请问有什么可以帮您？
结果: 您好 李四，请问有什么可以帮您？
```

---

## 5. 加入真实 LLM 调用

前面的例子用了硬编码的逻辑。现在加入真实的 LLM 调用来做意图分类。

```python
"""
lesson3_2_llm_graph.py
使用 LLM 做意图分类的 LangGraph
"""

from typing import TypedDict
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

# ========== 1. 定义 State ==========
class TriageState(TypedDict):
    question: str          # 用户问题
    intent: str            # 意图: billing / tech / support
    reply: str             # 回复内容

# ========== 2. 初始化 LLM ==========
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ========== 3. 定义 Nodes ==========
def classify_intent(state: TriageState) -> dict:
    """用 LLM 分类用户意图"""
    system_prompt = """你是一个意图分类助手。将用户问题分类为以下三类之一：
- billing: 账单、付款、发票、退款相关问题
- tech: 技术故障、功能使用、系统集成问题
- support: 其他客服问题

只返回类别名称（billing/tech/support），不要返回其他内容。"""

    messages = [
        SystemMessage(content=system_prompt),
        HumanMessage(content=state["question"]),
    ]

    response = llm.invoke(messages)
    intent = response.content.strip().lower()
    print(f"[Classify] 意图: {intent}")

    return {"intent": intent}

def handle_billing(state: TriageState) -> dict:
    """处理账单问题"""
    prompt = f"你是一个账单专家。请友好地回答用户的问题：{state['question']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"reply": response.content}

def handle_tech(state: TriageState) -> dict:
    """处理技术问题"""
    prompt = f"你是一个技术支持专家。请专业地回答用户的问题：{state['question']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"reply": response.content}

def handle_support(state: TriageState) -> dict:
    """处理一般问题"""
    prompt = f"你是一个友好的客服。请回答用户的问题：{state['question']}"
    response = llm.invoke([HumanMessage(content=prompt)])
    return {"reply": response.content}

# ========== 4. 路由函数 ==========
def route_by_intent(state: TriageState) -> str:
    """根据意图路由"""
    return state["intent"]

# ========== 5. 构建 Graph ==========
builder = StateGraph(TriageState)

builder.add_node("classify", classify_intent)
builder.add_node("billing", handle_billing)
builder.add_node("tech", handle_tech)
builder.add_node("support", handle_support)

builder.set_entry_point("classify")

builder.add_conditional_edges(
    "classify",
    route_by_intent,
    {"billing": "billing", "tech": "tech", "support": "support"}
)

builder.add_edge("billing", END)
builder.add_edge("tech", END)
builder.add_edge("support", END)

triage_graph = builder.compile()

# ========== 6. 运行 ==========
if __name__ == "__main__":
    test_questions = [
        "我这个月的账单怎么多了50块钱？",
        "你们的 App 一直闪退，怎么办？",
        "你们公司的上班时间是什么？",
    ]

    for q in test_questions:
        print(f"\n{'='*50}")
        print(f"用户: {q}")
        result = triage_graph.invoke({"question": q})
        print(f"回复: {result['reply']}")
```

---

## 6. 动手练习

### 练习 1：搭建一个审批流程图

实现一个简单的报销审批流程：

```
Start → 提交报销 → 判断金额 → (≤1000) → 自动通过 → End
                           → (>1000) → 经理审批 → End
```

提示：
1. State 需要包含：金额（amount）、审批结果（approved）
2. 需要一个路由函数根据金额判断走哪条路

<details>
<summary>参考代码框架</summary>

```python
from typing import TypedDict
from langgraph.graph import StateGraph, END

class ApprovalState(TypedDict):
    amount: float
    approved: bool
    message: str

def submit_node(state): ...
def auto_approve(state): ...
def manager_approve(state): ...

def route_by_amount(state):
    return "auto" if state["amount"] <= 1000 else "manager"

builder = StateGraph(ApprovalState)
builder.add_node("submit", submit_node)
builder.add_node("auto", auto_approve)
builder.add_node("manager", manager_approve)
builder.set_entry_point("submit")
builder.add_conditional_edges("submit", route_by_amount, {"auto": "auto", "manager": "manager"})
builder.add_edge("auto", END)
builder.add_edge("manager", END)
graph = builder.compile()
```

</details>

### 练习 2：给图添加状态日志

在每次节点执行前后打印当前状态，方便调试：

```python
def debug_node(state, name):
    """在节点执行前后打印状态"""
    print(f"[DEBUG] 进入 {name} 节点, 状态: {state}")
    # ... 执行节点逻辑 ...
    print(f"[DEBUG] 离开 {name} 节点")
```

---

## 7. 小结

本课要点：

- **State** 是在整个流程中传递的数据结构（TypedDict）
- **Node** 是接收 State、返回更新后 State 的普通函数
- **Edge** 表示固定的执行顺序
- **Conditional Edge** 通过路由函数根据 State 动态决定下一个节点
- 构建图的标准流程：定义 State → 定义 Nodes → 添加 Edges → 编译 → 运行
- 用 `get_graph().draw_mermaid()` 可以可视化你的流程图

**下节课预告**: 构建一个完整的多步骤决策系统 —— 客服分类 + 路由 + 处理 + 满意度循环。